package com.atguigu.chapter05.Transform;

import com.atguigu.chapter05.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

/**
 * TODO
 *
 * @author cjp
 * @version 1.0
 * @date 2021/3/3 15:29
 */
public class Flink11_Process {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

//        DataStreamSource<String> inputDS = env.readTextFile("input/sensor.csv");
        DataStreamSource<String> inputDS = env.socketTextStream("localhost", 9999);

        SingleOutputStreamOperator<WaterSensor> sensorDS = inputDS.map(new MapFunction<String, WaterSensor>() {
            @Override
            public WaterSensor map(String value) throws Exception {
                // 切分
                String[] line = value.split(",");
                return new WaterSensor(line[0], Long.valueOf(line[1]), Integer.valueOf(line[2]));

            }
        });

        // TODO Process'
        // keyby之前调用
//        sensorDS
//                .process(new ProcessFunction<WaterSensor, String>() {
//                    @Override
//                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
//                        out.collect(value.toString());
//                    }
//                })
//                .print("keyby之前");

        // keyby之后调用
        KeyedStream<WaterSensor, String> sensorKS = sensorDS.keyBy(sensor -> sensor.getId());
        sensorKS
                .process(new KeyedProcessFunction<String, WaterSensor, String>() {
                    @Override
                    public void processElement(WaterSensor value, Context ctx, Collector<String> out) throws Exception {
                        String currentKey = ctx.getCurrentKey();
                        out.collect(currentKey + "  ,  " + value.toString());
                    }
                })
                .print();


        env.execute();
    }
}
/*
  Reduce：
    输入和输出的类型必须一致，不够灵活
    分组内的第一条数据，不会调用 reduce 方法
    value1是之前的计算结果，value2是当前来的数据
 */